package com.climber.spb_rabbitmq.utils;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

public class RabbitMQUtils {
    private static final int MESSAGE_COUNT=20;

    public static Channel getChannel()  {
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");
            connectionFactory.setPort(5672);
            connectionFactory.setPassword("guest");
            connectionFactory.setUsername("guest");
            Connection connection = connectionFactory.newConnection();
            channel = connection.createChannel();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }

        return channel;
    }

    /**
     * 逐个确认
     */
    public static void publishMessageIndividually() {
        try {
            Channel channel = getChannel();
            channel.confirmSelect();
            channel.queueDeclare("individual queue", false, false, false, null);
//            send five messages
            long currentTimeMillis = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = i + "M";
                channel.basicPublish("", "individual queue", null, message.getBytes(StandardCharsets.UTF_8));
                boolean confirms = channel.waitForConfirms();
//                if(confirms){
//                    System.out.println("confirmed spend "+(System.currentTimeMillis()-currentTimeMillis));
//                }
            }
            System.out.println("all spend " + (System.currentTimeMillis() - currentTimeMillis));


        } catch (IOException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 批量确认
     */
    public static void publishMessageBatch() {
        try {
            Channel channel = getChannel();
            channel.confirmSelect();
            channel.queueDeclare("batchWait", false, false, false, null);
            int batchSize = 5;
            int unconfimNum = 0;
            long currentTimeMillis = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String mess = i + "M";
                channel.basicPublish("", "batchWait", null, mess.getBytes(StandardCharsets.UTF_8));
                unconfimNum++;
                if (unconfimNum == batchSize) {
                    channel.waitForConfirms();
                    unconfimNum = 0;
                }
            }
            System.out.println("all spend " + (System.currentTimeMillis() - currentTimeMillis));
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }


    /**
     * 异步确认
     */
    public static void publishMessageAsync() {
        try (
                Channel channel = RabbitMQUtils.getChannel()
        ) {
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName, false, false, false, null);
            //开启发布确认
            channel.confirmSelect();
            /**序号与消息进行关联 */
            ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
            /**
             * 确认收到消息的一个回调
             * 1.消息序列号
             * 2.true 可以确认小于等于当前序列号的消息  false 确认当前序列号消息
             */
            ConfirmCallback ackCallback = (sequenceNumber, multiple) -> {
                if (multiple) {
                    //返回的是小于等于当前序列号的未确认消息 是一个 map
                    // 返回确认的所有消息的序号，将这些序号从未确认的记录中删除
                    // 返回的map是对应数据集和 的映射，操作map可以直接操作 outstandingConfirms 中对应的那部分数据
                    ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(sequenceNumber, true);
                    confirmed.clear();//清除该部分未确认消息
                } else {
                    outstandingConfirms.remove(sequenceNumber);//只清除当前序列号的消息
                }
            };
            ConfirmCallback nackCallback = (sequenceNumber, multiple) ->
            {
                String message = outstandingConfirms.get(sequenceNumber);
                System.out.println("发布的消息" + message + "未被确认，序列号" + sequenceNumber);
            };
            /**
             * 添加一个异步确认的监听器
             * 1.确认收到消息的回调
             * 2.未收到消息的回调
             */
            channel.addConfirmListener(ackCallback, nackCallback);
            long begin = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = "消息" + i;
                outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
                channel.basicPublish("", queueName, null, message.getBytes());
            }
            long end = System.currentTimeMillis();
            System.out.println("发布" + MESSAGE_COUNT + "个异步确认消息,耗时" + (end - begin) +"ms");
        }catch (Exception e){
            e.printStackTrace();
        }

    }

    public static void dropQueue(String queueName){
        try {
            RabbitMQUtils.getChannel().queueDelete(queueName);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
//    dropQueue("05f8f671-829b-40b2-b73c-c3f00f4ee9b5");
        //        publishMessageIndividually(); //30
//        publishMessageBatch(); //13
//        publishMessageAsync(); // 11


    }
}
